package org.shj.hadoop.customerinputformat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

public class FindMaxValueInputFormat extends InputFormat<IntWritable, ArrayWritable>{

	public static float[] floatvalues;
	public RecordReader<IntWritable, ArrayWritable> createRecordReader(
			InputSplit splict, TaskAttemptContext context) throws IOException, InterruptedException{
		return new FindMaxValueRecordReader();
	}
	
	public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException{
		int numOfValues = context.getConfiguration().getInt("NumOfValues", 100);
		floatvalues = new float[numOfValues];
		Random ran = new Random();
		for(int i = 0 ; i < numOfValues; i++){
			floatvalues[i] = ran.nextFloat();
			
		}
		
		int numSplits = context.getConfiguration().getInt("mapred.map.tasks", 2);
		int begin = 0;
		int length = (int)Math.floor(numOfValues / numSplits);
		
		List<InputSplit> splits = new ArrayList<InputSplit>();
		int end = length -1;
		for(int i = 0 ; i < numSplits - 1 ; i++){
			FindMaxValueInputSplit split = new FindMaxValueInputSplit(begin, end);
			splits.add(split);
			System.out.println("FromTo: " + split.getM_StartIndex() +  "-->" + split.getM_EndIndex());
			begin = end + 1;
			end = begin + length - 1;
		}
		
		FindMaxValueInputSplit split = new FindMaxValueInputSplit(begin, numOfValues - 1);
		splits.add(split);
		
		return splits;
	}
}
